package org.example;


import com.zhaofujun.filequeue.*;

import java.io.IOException;
import java.time.Duration;
import java.time.LocalDateTime;

public class Test2 {
    public static void main(String[] args) throws IOException {
        KryoSerializerV2<String> stringKryoSerializerV2 = new KryoSerializerV2<>(String.class);
        FileQueue<String> fileQueueWithReactor = FileQueueFactory.createFileQueue(
                "queue",
                10000,
                stringKryoSerializerV2
        );


//        write(fileQueueWithReactor);
        writeWithThread(fileQueueWithReactor);


//        readWithThread(fileQueueWithReactor, 1);
        read(fileQueueWithReactor, 0);
//
//        System.in.read();
        fileQueueWithReactor.close();

    }

    private static void readWithThread(FileQueue<String> fileQueueWithReactor, int n) {
        // 模拟10个线程读取数据
        for (int i = 0; i < 10; i++) {
            new Thread(() -> {
                try {
                    while (true) {
                        String s = fileQueueWithReactor.pop(n);

                        if (s == null) {
                            try {
                                Thread.sleep(100);
                            } catch (InterruptedException e) {
                                throw new RuntimeException(e);
                            }
                        } else {
                            System.out.println("readWithThread:" + s);
                        }

                    }
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            }).start();
        }

    }

    private static void writeWithThread(FileQueue<String> fileQueueWithReactor) {
        // 模拟10个线程写入数据
        LocalDateTime startTime = LocalDateTime.now();
        Thread[] threads = new Thread[10];
        for (int i = 0; i < 10; i++) {
            threads[i] = new Thread(() -> {
                try {
                    for (int j = 0; j < 100000; j++) {
                        fileQueueWithReactor.push("123:" + Thread.currentThread().getName() + ":" + j);
                    }
                } catch (IOException e) {
                    throw new RuntimeException(e);
                }
            });
            threads[i].start();
        }
        for (int i = 0; i < 10; i++) {
            try {
                threads[i].join();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }

        LocalDateTime endTime = LocalDateTime.now();
        System.out.println("writeWithThread cost time:" + Duration.between(startTime, endTime).toMillis());
    }


    //
    private static void write(FileQueue<String> fileQueueWithReactor) throws IOException {
        LocalDateTime startTime = LocalDateTime.now();
        for (int i = 0; i < 1000000; i++) {
            fileQueueWithReactor.push("456:" + i);
        }
        LocalDateTime endTime = LocalDateTime.now();
        System.out.println("write cost time:" + Duration.between(startTime, endTime).toMillis());
    }

    private static void read(FileQueue<String> fileQueueWithReactor, int n) throws IOException {
        int i = 0;
        while (true) {
            String s = fileQueueWithReactor.pop(n);

            if (s == null) {
                System.out.println("read:" + i);
                return;
            } else {
                i++;
            }

        }
    }
}
